import * as URL from 'url'
import * as os from "os";
import * as path from "path";

import * as ObjectAssign from 'object-assign-merge'
import GetLogger from "wrapper-log4js";
import * as mq from 'amqplib'
import { Channel, Message } from 'amqplib'

import { EventBus, SubscribeOptions, MessageConsumeModel, PublishOptions, AMQPAPI } from "./EventBus";
import * as ChannelCacheManager from './internal/ChannelCacheManager';
import { EventBusConf, Host, Product } from './EventBusConf';

const config_path = path.resolve('./config/eventbus.json')
const config: EventBusConf = require(config_path);
const log = GetLogger('eventbus');
const conf: {
    hosts: { [key: string]: Host },
    products: { [key: string]: Product }
} = {
        hosts: {},
        products: {}
    };
const defaultSubscribeOptions: SubscribeOptions = {
    Model: MessageConsumeModel.Clustering,
    BasicQOS: 1,
    ShardChannel: true,
    FalilMessageReQueue: true
}
const defaultPublishOptions: PublishOptions = {
    DeliveryMode: true,
    ShardChannel: true
}
/**
 * 消息中心
 * @export
 * @class EventBus
 */
export class EventBus_AMQP implements EventBus, AMQPAPI {
    Cast<T>(): T {
        return <T><any>this;
    }
    Exit() {
        ChannelCacheManager.ClearChannel()
    }

    /**
     * 实例
     * 
     * @static
     * @type {EventBus}
     * @memberof EventBus
     */
    public static Instance = new EventBus_AMQP()

    private constructor() {

        log.Info('EventBus ctor')
        for (var index = 0; index < config.Hosts.length; index++) {
            var element = config.Hosts[index];
            conf.hosts[element.Name] = element;
        }

        for (var index = 0; index < config.Products.length; index++) {
            var element2 = config.Products[index];
            conf.products[element2.Id] = element2;
        }
    }

    /**
     * 发布消息
     * 
     * @template TModel 
     * @param {string} productId 产品名字
     * @param {string} topic exchange
     * @param {string} tag routekey
     * @param {string} id messageid
     * @param {TModel} message 
     * @param {PublishOptions} [options] 
     * @returns {Promise<boolean>} 发送成功返回true,否则返回false
     * @memberof EventBus
     */
    public Publish<TModel>(productId: string, topic: string, tag: string, id: string, message: TModel, options?: PublishOptions): Promise<boolean> {

        var _options: PublishOptions = ObjectAssign.assign({}, defaultPublishOptions, options)
        var bodyStr;
        var isStr = false;
        if (typeof message === 'string') {
            bodyStr = message
            isStr = true
        }
        else
            bodyStr = JSON.stringify(message)

        var body = Buffer.from(bodyStr)
        var url = this.GetAMQPUrl(productId)

        return new Promise((resolve, reject) => {
            ChannelCacheManager.GetChannel(url, topic, <boolean>_options.ShardChannel, this.CreateChannel).then(channel => {
                return channel.checkExchange(topic).then(() => { return channel })
            }).then(channel => {
                var pubResult = channel.publish(topic, tag, body, {
                    deliveryMode: _options.DeliveryMode,
                    messageId: id,
                    contentType: isStr ? 'text/plain' : 'application/json'
                })
                resolve(pubResult)
            }).catch(error => {
                log.Error('eventbus channel publish', error)
                resolve(false)
            })
        })


    }
    public Subscribe<TModel>(productId: string, topic: string, callback: (model: TModel, message: Message) => boolean | Promise<boolean>, options?: SubscribeOptions, ): Promise<any> {
        var _options: SubscribeOptions = ObjectAssign.assign({}, defaultSubscribeOptions, options)
        var url = this.GetAMQPUrl(productId)
        var queueName = topic

        return new Promise((resolve, reject) => {
            ChannelCacheManager.GetChannel(url, topic, <boolean>_options.ShardChannel, this.CreateChannel).then(channel => {

                if (_options.Model == MessageConsumeModel.Broadcasting) {
                    var ip = this.getIPAdress()
                    queueName = ip + '_' + productId + '_' + topic + Date.now()
                    return channel.assertQueue(queueName, { autoDelete: true }).then(() => {
                        return channel.bindQueue(queueName, topic, '')
                    }).then(() => {
                        return channel
                    })
                }

                return channel
            }).then(channel => {
                return channel.checkQueue(queueName).then(check => {
                    log.Info('sub check queue', check)
                    return channel
                })
            }).then(channel => {
                return channel.prefetch(<number>_options.BasicQOS, false).then(() => {
                    return channel
                })

            }).then(channel => {
                return channel.consume(queueName, message => {
                    try {
                        var json = (<Message>message).content.toString()
                        var first = json[0]
                        var last = json[json.length - 1];
                        var r: TModel;
                        if ((<Message>message).properties['contentType'] === 'application/json')
                            r = JSON.parse(json)
                        else
                            r = <any>json

                        var success = callback(r, (<Message>message))
                        channel.nack
                        if (typeof success === 'boolean') {
                            if (success)
                                channel.ack((<Message>message))
                            else {
                                log.Error('event bus consume fail. return false', { content: json, fields: (<Message>message).fields, properties: (<Message>message).properties })
                                channel.nack((<Message>message), false, _options.FalilMessageReQueue)
                            }
                        } else {
                            Promise.resolve(success).then(_success => {
                                if (_success)
                                    channel.ack((<Message>message))
                                else {
                                    log.Error('event bus consume fail. return promise false', { content: json, fields: (<Message>message).fields, properties: (<Message>message).properties })
                                    channel.nack((<Message>message), false, _options.FalilMessageReQueue)
                                }
                            }).catch(err => {
                                log.Error('event bus consume fail. return promise reject', err, { content: json, fields: (<Message>message).fields, properties: (<Message>message).properties })
                                channel.nack((<Message>message), false, _options.FalilMessageReQueue)
                            })
                        }

                    } catch (error) {
                        log.Error('event bus consume error', error, (<Message>message))
                        channel.nack(<Message>message, false, _options.FalilMessageReQueue)
                    }

                }, {
                        noAck: false
                    })
            }).catch(error => {
                log.Error('eventbus channel subscribe', error)
                resolve(false)
            })
        })

    }
    private getIPAdress() {
        var interfaces = require('os').networkInterfaces();
        for (var devName in interfaces) {
            var iface = interfaces[devName];
            for (var i = 0; i < iface.length; i++) {
                var alias = iface[i];
                if (alias.family === 'IPv4' && alias.address !== '127.0.0.1' && !alias.internal) {
                    return alias.address;
                }
            }
        }
    }
    private GetAMQPUrl(productId: string) {
        var prod = conf.products[productId]
        if (!prod)
            throw new Error('productId:"' + productId + '" not in config/eventbus.json')

        var host = conf.hosts[prod.Host];
        if (!host)
            throw new Error('host:"' + prod.Host + '" not in config/eventbus.json')

        var url = URL.format({
            protocol: 'amqp',
            slashes: true,
            hostname: host.Address,
            port: host.Port,
            pathname: host.VHost,
            auth: host.UserName + ':' + host.UserPassword
        })
        return url
    }
    private CreateChannel(url: string, topic: string, cacheKey: string): Promise<Channel> {
        log.Info('eventbus create channel：：：' + url + '       cacheKey:' + cacheKey)
        return mq.connect(url).then(conn => {
            return conn.createChannel()
        }).then(channel => {
            // channel.setMaxListeners(20)
            channel.on('close', () => {
                log.Info(cacheKey + ' eventbus channel close')
            })

            channel.on('error', error => {
                log.Error(cacheKey + ' eventbus channel error', error)
            })

            channel.on('return', error => {
                log.Info(cacheKey + ' eventbus channel return', error)
            })

            channel.on('drain', error => {
                log.Info(cacheKey + ' eventbus channel drain', error)
            })
            return channel
        })

    }

    public SendToQueue(productId: string, queueName: string, message: Buffer, properties: any, options?: PublishOptions): Promise<boolean> {
        var _options: PublishOptions = ObjectAssign.assign({}, defaultPublishOptions, options)

        var url = this.GetAMQPUrl(productId)
        return ChannelCacheManager.GetChannel(url, queueName, <boolean>_options.ShardChannel, this.CreateChannel).then(channel => {
            return channel.checkQueue(queueName).then(() => { return channel })
        }).then(channel => {
            return channel.sendToQueue(queueName, message, properties)
        })
    }
}



// export interface EventBus {
//     Publish<TModel>(productId: string, topic: string, tag: string, id: string, message: TModel, options?: PublishOptions): Promise<boolean>;
//     Subscribe<TModel>(productId: string, topic: string, callback: (model: TModel) => boolean, options?: SubscribeOptions, ): Promise<any>;
// }